﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using ZooKeeperNet;
using Org.Apache.Zookeeper.Data;
using ZK = ZooKeeperNet;

using WorkStation.Frameworks.Core;
using WorkStation.Frameworks.Core.Extensions;
using WorkStation.Frameworks.EasyBus.DbModel;
using WorkStation.Frameworks.EasyBus.Handles;
using WorkStation.Frameworks.EasyBus.Pipeline;
using WorkStation.Frameworks.EasyBus.Pipeline.Extensions;
using WorkStation.Frameworks.EasyBus.Pipeline.Handles;

namespace WorkStation.Frameworks.EasyBus.Repositories.ZooKeeper
{
    /// <summary>
    /// 订阅端变更监视器
    /// </summary>
    class SubscribeInfoChangeWatcher : IWatcher
    {
        protected IBus _bus = null;
        protected ZK.ZooKeeper _zk;

        public SubscribeInfoChangeWatcher(IBus bus, ZK.ZooKeeper zk) 
        {
            this._bus = bus;
            this._zk = zk;
        }

        public void Process(WatchedEvent @event)
        {
            // 断开连接和会话过期不做处理
            switch (@event.State)
            { 
                case KeeperState.Disconnected:
                case KeeperState.Expired:
                    return;
            }

            Stat stat = this._zk.Exists(@event.Path, false);

            // 订阅配置变更事件
            if (@event.Type == EventType.NodeDataChanged)
            {
                SubscribeInfo info = this._zk.GetData(@event.Path, new SubscribeInfoChangeWatcher(this._bus, this._zk), stat).ZooNodeToObject<SubscribeInfo>();

                this._bus.Publish(new SubscribeInfoChangeEvent() { Change = info, SubscribeInfoChangeType = SubscribeInfoChangeTypeEnum.Change });
            }

            // 订阅配置失效事件
            if (@event.Type == EventType.NodeDeleted) 
            {
                string[] paths = @event.Path.Split('/');

                string[] args = paths.Last().Split('-');

                SubscribeInfo info = new SubscribeInfo() 
                {
                    Name = args[2],
                    RouteKey = paths[4],
                    NodeName = args[1],
                    BusName = args[0]
                };
                
                this._bus.Publish(new SubscribeInfoChangeEvent() { Change = info, SubscribeInfoChangeType = SubscribeInfoChangeTypeEnum.OffLine });
            }

            // ZooKeeper 观察订阅端 BusName 节点（用来观察是不是新增节点）
            if (@event.Type == EventType.NodeChildrenChanged) 
            {
                // @event.Path is RouteKeyNode path

                IBusDataManager manager = this._bus as IBusDataManager;

                foreach(string subscribeName in this._zk.GetChildren(@event.Path, new SubscribeInfoChangeWatcher(this._bus, this._zk)))
                {
                    string subscribeNameNodePath = "{0}/{1}".Fmt(@event.Path, subscribeName);

                    Stat subscribeNameStat = this._zk.Exists(subscribeNameNodePath, false);

                    (subscribeNameStat != null).WhenTrue(() => 
                    {
                        SubscribeInfo info = this._zk.GetData(subscribeNameNodePath, new SubscribeInfoChangeWatcher(this._bus, this._zk), subscribeNameStat).Change(data => data.ZooNodeToObject<SubscribeInfo>());

                        if (info != null)
                        {
                            IHandle handle = manager.SendMessagePipeline.GetHandle(info.RouteKey, DefaultPublishHandle.GetHandleName(info));

                            if (handle == null)
                            {
                                // 新增
                                this._bus.Publish(new SubscribeInfoChangeEvent() { Change = info, SubscribeInfoChangeType = SubscribeInfoChangeTypeEnum.OnLine });
                            }
                        }
                    });
                }
            }
        }
    }
}
